package com.ferry.config.client.monitor;

import com.ferry.config.client.listener.ConfigChangeListener;
import com.ferry.config.client.listener.GlobalConfigChangeListener;
import com.ferry.config.client.loader.HttpGetClassLoader;
import com.ferry.config.client.loader.LongPollConfigLoader;
import com.ferry.config.client.vo.ConfigRepository;
import com.ferry.config.client.vo.GetData;
import com.google.common.collect.MapDifference;
import com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by daiyong
 */
public class ConfigMonitor {
	private static final Logger LOGGER = LoggerFactory.getLogger(ConfigMonitor.class);

	private String nameSpace;

	private String env;

	private static final Object lock = new Object();

	private static final String SCAN_BASE_PACKAGE = ".";

	/**
	 * 内部监听器
	 */
	private ConcurrentMap<String, Set<ConfigChangeListener>> innerListeners = new ConcurrentHashMap<String, Set<ConfigChangeListener>>();

	/**
	 * 外部监听器
	 */
	private ConcurrentMap<String, Set<ConfigChangeListener>> outerListeners = new ConcurrentHashMap<String, Set<ConfigChangeListener>>();

	/**
	 * 全局监听器
	 */
	private volatile GlobalConfigChangeListener globalConfigChangeListener = null;
	//数据更新线程
	private ExecutorService configListenerExecutor = Executors.newFixedThreadPool(1);
	//数据更新线程
	private ExecutorService globalConfigListenerExecutor = Executors.newFixedThreadPool(1);

	/**
	 * 构造器
	 * @param nameSpace
	 * @param env
	 */
	public ConfigMonitor(String nameSpace, String env) {
		this.nameSpace = nameSpace;
		this.env = env;
	}

	/**
	 * 本地缓存
	 */
	private ConfigRepository configRepository = new ConfigRepository();

	public void init() {

		//每10s 全量拉取一次
		new HttpGetClassLoader(this.nameSpace, this.env, this).start();

//		先启动全量拉取loader
		new LongPollConfigLoader(this.nameSpace, this.env, this).start();

		LOGGER.info("ConfigMonitor init success!");
	}

	public void addListener(String key, ConfigChangeListener listener) {
		if (!outerListeners.containsKey(key)) {
			synchronized (lock) {
				if (!outerListeners.containsKey(key)) {
					outerListeners.put(key, new HashSet<ConfigChangeListener>());
				}
			}
		}
		outerListeners.get(key).add(listener);
	}

	public void removeListener(String key, ConfigChangeListener listener) {
		if (outerListeners.containsKey(key)) {
			synchronized (lock) {
				if (outerListeners.containsKey(key)) {
					Set<ConfigChangeListener> listeners = outerListeners.get(key);
					if (null != listeners && !listeners.isEmpty()) {
						outerListeners.get(key).remove(listener);
					}
				}
			}
		}
	}

	public String getValue(String key) {
		return configRepository.getConfig().get(key);
	}

	public Map<String, String> getAllKeyValues() {
		return new HashMap<String, String>(configRepository.getConfig());
	}

	public Set<String> getAllKeys() {
		return new HashSet<String>(configRepository.getConfig().keySet());
	}

	/**
	 * 填充本地缓存
	 * @param configs
	 */
	public void fillConfigRepository(List<GetData> getDatas) {

		if (!CollectionUtils.isEmpty(getDatas)) {

			ConcurrentMap<String, String> newDiffData = new ConcurrentHashMap<String, String>();
			long maxUtime = configRepository.getMaxUtime();
			for (GetData getData : getDatas) {
				newDiffData.put(getData.getKey(), getData.getContent());
				maxUtime = Math.max(maxUtime, getData.getUtime());
			}

			//更新最大更新时间
			configRepository.setMaxUtime(maxUtime);

			ConcurrentMap<String, String> oldData = configRepository.getConfig();

			ConcurrentMap<String, String> oldDataCopy = copyMap(oldData);

			for (Map.Entry<String, String> entry : newDiffData.entrySet()) {
				String key = entry.getKey();
				String value = entry.getValue();
				oldData.put(key, value);
			}

			notifyGlobalConfigListener(oldDataCopy, oldData);

			diff(oldDataCopy, oldData);
		}

	}


	private void diff(final Map<String, String> oldData, final Map<String, String> newData) {
		if (oldData == null || newData == null) {
			return;
		}
		MapDifference<String, String> diffData = Maps.difference(oldData, newData);
		configDeleted(diffData);
		configUpdated(diffData);
		configAdded(diffData);
	}

	private void configAdded(MapDifference<String, String> diffData) {
		for (Map.Entry<String, String> addedEntry : diffData.entriesOnlyOnRight().entrySet()) {
			configChanged(addedEntry.getKey(), null, addedEntry.getValue());
		}
	}

	private void configUpdated(MapDifference<String, String> diffData) {
		for (Map.Entry<String, MapDifference.ValueDifference<String>> changedEntry : diffData.entriesDiffering().entrySet()) {
			MapDifference.ValueDifference<String> oldAndNewValue = changedEntry.getValue();
			configChanged(changedEntry.getKey(), oldAndNewValue.leftValue(), oldAndNewValue.rightValue());
		}
	}

	private void configDeleted(MapDifference<String, String> diffData) {
		for (Map.Entry<String, String> deletedEntry : diffData.entriesOnlyOnLeft().entrySet()) {
			configChanged(deletedEntry.getKey(), deletedEntry.getValue(), null);
		}
	}

	private void configChanged(final String key, final String oldValue, final String newValue) {
		configListenerExecutor.execute(new Runnable() {
			@Override
			public void run() {
				if (outerListeners.containsKey(key)) {
					executeOuterListener(key, oldValue, newValue);
				}
			}
		});
	}

	private void executeOuterListener(final String key, final String oldValue, final String newValue) {
		Set<ConfigChangeListener> configChangeListeners = outerListeners.get(key);
		if (!CollectionUtils.isEmpty(configChangeListeners)) {
			for (final ConfigChangeListener listener : configChangeListeners) {
				listener.changed(key, oldValue, newValue);
			}
		}
	}


	/**
	 * 通知全局监听器
	 * @param oldData
	 * @param newData
	 */
	public void notifyGlobalConfigListener(final ConcurrentMap<String, String> oldData, final ConcurrentMap<String, String> newData) {
		globalConfigListenerExecutor.execute(new Runnable() {
			@Override
			public void run() {
				if (null != globalConfigChangeListener) {
					globalConfigChangeListener.changed(oldData, newData);
				}
			}
		});
	}

	/**
	 * 单独key listener
	 * @param key
	 * @param oldValue
	 * @param newValue
	 */
	public void notifyConfigListener(final String key, final String oldValue, final String newValue) {

	}

	private ConcurrentMap<String, String> copyMap(ConcurrentMap<String, String> oldData) {
		ConcurrentMap<String, String> newData = new ConcurrentHashMap<String, String>();
		if (null != oldData) {
			for (Map.Entry<String, String> entry : oldData.entrySet()) {
				newData.put(entry.getKey(), entry.getValue());
			}
		}
		return newData;
	}

	public void setGlobalConfigChangeListener(GlobalConfigChangeListener globalConfigChangeListener) {
		this.globalConfigChangeListener = globalConfigChangeListener;
	}


	public void destroy() {

	}

	public long getConfigMaxUtime() {
		return configRepository.getMaxUtime();
	}
}
